package com.yushixin.order.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.yushixin.order.model.Account;
import com.yushixin.common.constant.Constant;
import com.yushixin.order.repository.IAccountRepository;
import com.yushixin.order.service.MQConsumerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.Optional;

/**
 * MQ消费者实现类
 */
@Component
@Slf4j
public class MQConsumerServiceImpl implements MQConsumerService {
    
    @Value("${project.time.delay-double-delete}")
    private Long delayDoubleDeleteTime;
    private static Long delayDoubleDeleteTimeCopy;
    
    @Autowired
    private IAccountRepository accountRepository;
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    /**
     * 初始化方法
     */
    @PostConstruct
    public void init() {
        log.info("MQ消费者初始化开始");
        // Redis缓存延时双删间隔时间
        delayDoubleDeleteTimeCopy = delayDoubleDeleteTime;
        // Account表本地数据加载到Redis中
        List<Account> accounts = accountRepository.findAll();
        if (!CollectionUtils.isEmpty(accounts)) {
            ValueOperations<String, Object> ops = redisTemplate.opsForValue();
            accounts.forEach((account) -> ops.set(account.getId().toString(), account));
        }
        log.info("MQ消费者初始化完成");
    }
    
    /**
     * MQ新增消息
     */
    @Service
    @RocketMQMessageListener(topic = Constant.MQTopic, selectorExpression = Constant.MQTagAdd, consumerGroup = Constant.MQGroupAdd)
    public static class ConsumerAdd implements RocketMQListener<String> {
        @Autowired
        private IAccountRepository accountRepository;
        @Autowired
        private RedisTemplate<String, Object> redisTemplate;
        
        @Override
        public void onMessage(String message) {
            log.info("消费者,接收新增消息:[{}]", message);
            // 解析JSON
            Account account = JSONObject.parseObject(message, Account.class);
            // 新增数据库
            accountRepository.save(account);
            // 新增缓存
            redisTemplate.opsForValue().set(account.getId().toString(), account);
        }
    }
    
    /**
     * MQ修改消息
     */
    @Service
    @RocketMQMessageListener(topic = Constant.MQTopic, selectorExpression = Constant.MQTagUpdate, consumerGroup = Constant.MQGroupUpdate)
    public static class ConsumerUpdate implements RocketMQListener<String> {
        @Autowired
        private IAccountRepository accountRepository;
        @Autowired
        private RedisTemplate<String, Object> redisTemplate;
        
        @Override
        public void onMessage(String message) {
            log.info("消费者,接收修改消息:[{}]", message);
            // 解析JSON
            Account account = JSONObject.parseObject(message, Account.class);
            Long id = account.getId();
            // 按账户ID查询
            Optional<Account> optional = accountRepository.findById(id);
            if (optional.isPresent()) {
                Account value = optional.get();
                BeanUtils.copyProperties(account, value);
                // 修改数据库
                accountRepository.save(value);
                // 执行延时双删策略
                MQConsumerServiceImpl.delayDoubleDelete(redisTemplate, id);
                // 新增缓存
                redisTemplate.opsForValue().set(id.toString(), account);
            } else {
                log.error("修改记录,记录不存在, id:[{}]", id);
            }
        }
    }
    
    /**
     * MQ删除消息
     */
    @Service
    @RocketMQMessageListener(topic = Constant.MQTopic, selectorExpression = Constant.MQTagDelete, consumerGroup = Constant.MQGroupDelete)
    public static class ConsumerDelete implements RocketMQListener<String> {
        @Autowired
        private IAccountRepository accountRepository;
        @Autowired
        private RedisTemplate<String, Object> redisTemplate;
        
        @Override
        public void onMessage(String message) {
            log.info("消费者,接收删除消息:[{}]", message);
            // 解析JSON
            Account account = JSONObject.parseObject(message, Account.class);
            Long id = account.getId();
            // 按账户ID查询
            Optional<Account> optional = accountRepository.findById(id);
            if (optional.isPresent()) {
                // 删除数据库
                accountRepository.deleteById(id);
                // 执行延时双删策略
                MQConsumerServiceImpl.delayDoubleDelete(redisTemplate, id);
            } else {
                log.error("删除记录,记录不存在, id:[{}]", id);
            }
        }
    }
    
    /**
     * Redis缓存延时双删
     * 1.延时200ms后再次删除
     * @param id 账户ID(缓存Key)
     */
    public static void delayDoubleDelete(final RedisTemplate<String, Object> redisTemplate, Long id) {
        // 缓存Key
        String key = id.toString();
        // 判断缓存是否存在
        if (!Boolean.TRUE.equals(redisTemplate.hasKey(key))) {
            log.warn("缓存中不存在Key:[{}],跳过处理", key);
            return;
        }
        // 删除缓存
        Boolean isDelete = redisTemplate.delete(key);
        // 删除缓存失败,延时双删
        if (!Boolean.TRUE.equals(isDelete)) {
            log.warn("消费者,删除缓存失败,执行延时双删策略,ID:[{}]", key);
            try {
                Thread.sleep(delayDoubleDeleteTimeCopy);
                isDelete = redisTemplate.delete(key);
                if (!Boolean.TRUE.equals(isDelete)) {
                    log.error("消费者,延时双删执行失败,请手动处理,ID:[{}]", key);
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }
}